﻿using ServerCommon.Log;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Net.WebSockets;
using System.Text;
using System.Threading.Tasks;

namespace ServerCommon.Net
{
    public enum EnumSendProtocolReturn
    {
        OK = 0,
        Error = -1,
        Invalid_Param = -2,
        Invalid_Node = -3,
        Invalid_Session = -4,
        Invalid_Zone = -5,
        Invalid_Socket = -6,
        Has_Error = -7
    }

    public class NetNode
    {
        //日志
        private static LogHelper log = LogHelper.GetLogger(typeof(NetClient));

        //io事件
        protected ConcurrentStack<SocketAsyncEventArgs> _ioSaeaPool;

        //缓存管理器
        protected BufferManager _bufferManager;

        public NetNodeManager Manager => _manager;
        protected NetNodeManager _manager;

        //client是远端链接点，service是绑定本地listen
        public IPEndPoint Addr => _addr;
        protected IPEndPoint _addr;

        public Guid HashCode => _hashCode;
        protected readonly Guid _hashCode;

        public long NodeId => _nodeId;
        protected long _nodeId;    //作为client用时，是远端的进程serverId，作为service用时，是自己进程的serverId

        public EnumNodeNetType NetType => _netType;
        protected EnumNodeNetType _netType;

        public EnumNodeServiceType ServiceType => _serviceType;
        protected EnumNodeServiceType _serviceType;

        //session的guid为key
        protected Dictionary<Guid, NetSession> dictSession = new Dictionary<Guid, NetSession>();

        //远端进程的id为key
        protected Dictionary<long, NetSession> dictIdSession = new Dictionary<long, NetSession>();

        //包体长度，4字节的int值
        public static readonly int MessageHeadLength = 4;

        public NetNode()
        {
            _hashCode = Guid.NewGuid();
            _manager = null;

            _netType = EnumNodeNetType.Error;
            _serviceType = EnumNodeServiceType.Error;
            _nodeId = 0;
        }

        protected void Config(NetNodeManager manager)
        {
            _manager = manager;
        }

        protected void CreateIOSAEAPool(int IOSAEANum)
        {
            for (int i = 0; i < IOSAEANum; i++)
            {
                _ioSaeaPool.Push(CreateIOEventArgs());
            }
        }

        protected SocketAsyncEventArgs CreateIOEventArgs()
        {
            SocketAsyncEventArgs ioEventArgs = new SocketAsyncEventArgs();
            _bufferManager.SetBuffer(ioEventArgs);
            ioEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
            ioEventArgs.UserToken = new DataToken() { BufferOffset = ioEventArgs.Offset };

            return ioEventArgs;
        }
        public virtual void Start()
        {
        }

        public virtual void Stop()
        {
        }

        public void CloseClientSocket(NetSession session)
        {
            _manager.PushIoEvent(new ConnectionEventArgs()
            {
                EventType = EnumConnectionEventType.Close,
                Session = session,
                NodeId = _nodeId,
            });

            try
            {
                session?.Close();
                OnDisConnectSession(session);
            }
            catch (Exception ex)
            {
                log.ErrorFormat("CloseClientSocket error: {0}", ex.Message);
            }
        }

        public void ShowPrompt()
        {
            log.Info(GetPrompt());
        }

        public string GetPrompt()
        {
            return string.Format("{{{0}}}", ToLog().ToString());
        }

        public virtual StringBuilder ToLog()
        {
            return new StringBuilder()
                .AppendFormat("NetType:{0},", NetType)
                .AppendFormat("ServiceType:{0},", ServiceType)
                .AppendFormat("Addr:{0},", Addr)
                .AppendFormat("NodeId:{0}", NodeId);
        }

        /// <summary>
        /// 在Task线程中发送数据
        /// </summary>
        /// <param name="clientSocket"></param>
        /// <param name="buffer"></param>
        public bool SendMessageAsync(NetSession session, byte[] buffer)
        {
            if (null == session)
            {
                log.WarnFormat("SendMessageAsync failed: clientSocket == null nodetype{0}", _serviceType);
                return false;
            }

            lock (session.SocketSyncLock)
            {
                session.Write(buffer);
                if (!session.TrySetSendFlag())
                {
                    return true;
                }
            }

            Task.Run(() =>
            {
                SocketAsyncEventArgs ioEvent = null;
                try
                {
                    //理论上永远不应该打印下面这句 ！！！！
                    if (!_ioSaeaPool.TryPop(out ioEvent))
                    {
                        log.Error($"SendMessageAsync failed. ioEvent use up. oops!!! node type: {_serviceType}");
                        session.ResetSendFlag();
                        return false;
                    }
                    ioEvent.AcceptSocket = session.WorkSocket;

                    TryDequeueAndPostSend(session, ioEvent);
                }
                catch (Exception ex)
                {
                    CloseClientSocket(ioEvent);
                    session.ResetSendFlag();
                    log.ErrorFormat("SendMessageAsync failed IP: {0} Hashcode:{1} error: {2}", session.RemoteEndPoint.ToString(), session.HashCode, ex.Message);
                }
                return true;
            });
            return true;
        }

        //
        public void CloseClientSocket(SocketAsyncEventArgs ioEventArgs)
        {
            var dataToken = (DataToken)ioEventArgs.UserToken;

            _manager.PushIoEvent(new ConnectionEventArgs()
            {
                EventType = EnumConnectionEventType.Close,
                Session = dataToken.Session,
                NodeId = _nodeId,
            });

            try
            {
                if (ioEventArgs.AcceptSocket != null)
                {
                    if(ioEventArgs.AcceptSocket.Connected)
                    {
                        ioEventArgs.AcceptSocket?.Shutdown(SocketShutdown.Both);
                    }
                    ioEventArgs.AcceptSocket?.Close();
                }

                ReleaseIOEventArgs(ioEventArgs);
                OnDisConnectSession(dataToken.Session);

            }
            catch (Exception e)
            {
                log.ErrorFormat("CloseClientSocket :{0}", e.Message);
            }
        }

        public void AddNetSession(long remoteNodeId, NetSession session)
        {
            if(dictIdSession.ContainsKey(remoteNodeId))
            {
                log.ErrorFormat("AddNetSession sessionId:{0} exist!!!", remoteNodeId);
            }
            else
            {
                dictIdSession.Add(remoteNodeId, session);
            }
        }

        protected virtual void OnDisConnectSession(NetSession session)
        {
            dictSession.Remove(session.HashCode);
            dictIdSession.Remove(session._remoteNodeId);
        }

        public int GetBufferSizeForEachBlock()
        {
            return _bufferManager.BufferSizeForEachBlock;
        }

        protected void ReleaseIOEventArgs(SocketAsyncEventArgs ioEventArgs)
        {
            if (ioEventArgs == null)
            {
                return;
            }
            var dataToken = (DataToken)ioEventArgs.UserToken;
            dataToken?.Reset(true);
            ioEventArgs.AcceptSocket = null;
            ioEventArgs.SetBuffer(dataToken.BufferOffset, GetBufferSizeForEachBlock());
            _ioSaeaPool.Push(ioEventArgs);
        }

        protected void ResetSAEAObject(SocketAsyncEventArgs saea)
        {
            try
            {
                saea.AcceptSocket?.Close();
            }
            catch (Exception ex)
            {
                log.ErrorFormat("ResetSAEAObject error: {0}", ex.Message);
            }
            saea.AcceptSocket = null;
        }

        protected void IO_Completed(object sender, SocketAsyncEventArgs ioEventArgs)
        {
            DataToken dataToken = (DataToken)ioEventArgs.UserToken;
            try
            {
                switch (ioEventArgs.LastOperation)
                {
                    case SocketAsyncOperation.Receive:
                        ProcessReceive(ioEventArgs);
                        break;
                    case SocketAsyncOperation.Send:
                        ProcessSend(ioEventArgs);
                        break;
                    default:
                        throw new ArgumentException(string.Format("LastOperation is not send or receive: {0}", ioEventArgs.LastOperation));
                }
            }
            catch (ObjectDisposedException)
            {
                CloseClientSocket(ioEventArgs);
            }
            catch (Exception ex)
            {
                log.ErrorFormat("IO_Completed Failed : {0}", ex.Message);
            }
        }

        protected void PostReceive(SocketAsyncEventArgs ioEventArgs)
        {
            if (ioEventArgs.AcceptSocket == null)
            {
                return;
            }
            try
            {
                bool willRaiseEvent = ioEventArgs.AcceptSocket.ReceiveAsync(ioEventArgs);

                if (!willRaiseEvent)
                {
                    ProcessReceive(ioEventArgs);
                }
            }
            catch (Exception ex)
            {
                log.ErrorFormat("PostReceive error: {0}", ex.Message);
            }
        }

        protected void ProcessReceive(SocketAsyncEventArgs ioEventArgs)
        {
            DataToken dataToken = ioEventArgs.UserToken as DataToken;
            if (dataToken == null)
            {
                CloseClientSocket(ioEventArgs);
                return;
            }

            if (ioEventArgs.SocketError != SocketError.Success)
            {
                _manager.PushIoEvent(new ConnectionEventArgs()
                {
                    EventType = EnumConnectionEventType.Disconnect,
                    Session = dataToken.Session,
                    NodeId = _nodeId,
                });

                //TODO
                log.WarnFormat("ProcessReceive disconnect IP:{0} ,socket error:{1} , bytes len: {2}",
                    dataToken.Session.RemoteEndPoint.ToString(), ioEventArgs.SocketError.ToString(), ioEventArgs.BytesTransferred);
                CloseClientSocket(ioEventArgs);
                return;
            }

            if (ioEventArgs.BytesTransferred == 0)
            {
                _manager.PushIoEvent(new ConnectionEventArgs()
                {
                    EventType = EnumConnectionEventType.Disconnect,
                    Session = dataToken.Session,
                    NodeId = _nodeId,
                });
                //对方主动关闭
                log.WarnFormat("ProcessReceive disconnect IP:{0} socket error:{1}", dataToken != null ? dataToken.Session.RemoteEndPoint.ToString() : " ", ioEventArgs.SocketError);
                CloseClientSocket(ioEventArgs);
                return;
            }

            //这里读取socket上的数据并处理
            var ret = TryReceiveMessage(ioEventArgs);

            if (ret != EnumErrorCode.Success)
            {
                log.WarnFormat("ProcessReceive disconnect IP:{0} TryReceiveMessage error:{1}", dataToken.Session.RemoteEndPoint.ToString(), ret);
                CloseClientSocket(ioEventArgs);
                return;
            }
            PostReceive(ioEventArgs);
        }

        /// <summary>
        /// 解析网络接收的数据
        /// </summary>
        /// <param name="ioEventArgs"></param>
        /// <returns></returns>
        public EnumErrorCode TryReceiveMessage(SocketAsyncEventArgs ioEventArgs)
        {
            var res = EnumErrorCode.Success;
            do
            {
                try
                {
                    if (ioEventArgs.UserToken is not DataToken)
                    {
                        log.Error("RequestHandler.TryReceiveMessage dataToken == null}");
                        break;
                    }
                    res = TryReadMessage(ioEventArgs);
                }
                catch (Exception ex)
                {
                    log.Error($"TryReceiveMessage: {ToLog()}", ex);
                }
            } while (false);

            return res;
        }

        /// <summary>
        /// 每次都拆分整包，残包复制到outIOEvents中，outIOEvents的接收从残包的offset开始
        /// </summary>
        /// <param name="ioEventArgs"></param>
        /// <returns></returns>
        public virtual EnumErrorCode TryReadMessage(SocketAsyncEventArgs ioEventArgs)
        {
            if (null == ioEventArgs)
            {
                log.Error("TryReadMessage null == ioEventArgs");
                return EnumErrorCode.IOEventArgsIsNull;
            }
            var buffer = ioEventArgs.Buffer;
            var dataToken = ioEventArgs.UserToken as DataToken;
            if (dataToken == null)
            {
                return EnumErrorCode.DataTokenIsNull;
            }
            var session = dataToken.Session;
            //当次需要处理的数据总大小= 当次接收到的数据大小 + 残包大小 
            int currentTotalCount = ioEventArgs.BytesTransferred + dataToken.LastDataCount;
            int remainingCount = currentTotalCount;
            var bufferBlockSize = GetBufferSizeForEachBlock();

            // 当前buffer已经读取到哪里
            var currentDataIndex = 0;
            while (remainingCount > MessageHeadLength)
            {
                // 计算下一个包的长度
                var nextProtocolDataLen = BitConverter.ToInt32(ioEventArgs.Buffer, dataToken.BufferOffset + currentDataIndex);
                //如果大于缓存的长度
                if (nextProtocolDataLen > bufferBlockSize)
                {
                    log.ErrorFormat("TryReadMessage remainingCount: {0} fullPackageLen: {1} currentDataIndex: {2} TotalCount: {3} dataToken.BufferOffset: {4} MessageHeadLength :{5} ,ioEventArgs.Offset :{6} NetNodeType: {7}",
                        remainingCount, nextProtocolDataLen + MessageHeadLength, currentDataIndex, currentTotalCount, dataToken.BufferOffset, MessageHeadLength, ioEventArgs.Offset, _netType);
                    return EnumErrorCode.OverMaxMessageLen;
                }
                if (nextProtocolDataLen <= 0)
                {
                    log.ErrorFormat("TryReadMessage remainingCount: {0} fullPackageLen: {1} currentDataIndex: {2} TotalCount: {3} dataToken.BufferOffset: {4} MessageHeadLength :{5} ,ioEventArgs.Offset :{6} NetNodeType: {7}",
                        remainingCount, nextProtocolDataLen + MessageHeadLength, currentDataIndex, currentTotalCount, dataToken.BufferOffset, MessageHeadLength, ioEventArgs.Offset, _netType);
                    return EnumErrorCode.LessZero;
                }
                // 如果数据有一个完整包的长度，直接解析为协议
                if (nextProtocolDataLen <= remainingCount - MessageHeadLength)
                {
                    currentDataIndex = currentDataIndex + MessageHeadLength;
                    var protocolBytes = new byte[nextProtocolDataLen];
                    Buffer.BlockCopy(ioEventArgs.Buffer, dataToken.BufferOffset + currentDataIndex, protocolBytes, 0, nextProtocolDataLen);
                    //TODO解析协议
                    {
                        bool onProcessMesageRet = OnProcessMessage(session, protocolBytes);
                        if (!onProcessMesageRet)
                        {
                            return EnumErrorCode.FaildProcessMessage;
                        }
                    }
                    currentDataIndex = currentDataIndex + nextProtocolDataLen;
                    remainingCount = currentTotalCount - currentDataIndex;
                }
                else
                {
                    break;
                }
            }

            dataToken.LastDataCount = remainingCount;
            // 如果还有残包，将残包数据复制到新的ioEvent上
            if (remainingCount == 0)
            {
                ioEventArgs.SetBuffer(dataToken.BufferOffset, bufferBlockSize);
                return EnumErrorCode.Success;
            }

            Buffer.BlockCopy(ioEventArgs.Buffer, dataToken.BufferOffset + currentDataIndex, ioEventArgs.Buffer, dataToken.BufferOffset, remainingCount);
            ioEventArgs.SetBuffer(dataToken.BufferOffset + remainingCount, bufferBlockSize - remainingCount);

            return EnumErrorCode.Success;
        }

        /// <summary>
        /// 数据的反序列化在线程中进行，然后再放到主线程处理协议
        /// </summary>
        /// <param name="exsocket"></param>
        /// <param name="data"></param>
        /// <returns></returns>
        public virtual bool OnProcessMessage(NetSession session, byte[] data)
        {
            if (null == session || null == data)
            {
                return false;
            }

            var messageEvent = new ConnectionEventArgs
            {
                EventType = EnumConnectionEventType.Recv,
                Message = new DataMessage { Data = data },
                Session = session,
                NodeId = _nodeId,
            };

            int protoId = System.BitConverter.ToInt32(data, 0);
            var parse = ProtocolList.GetParser(_serviceType, protoId);
            if (parse != null)
            {
                var p = parse.ParseFrom(data, 4, data.Length - 4) as Pb.IProtocol;
                if (p != null)
                {
                    messageEvent.RecvedProtocol = p;
                }
                else
                {
                    log.ErrorFormat("OnProcessMessage parse failed!");
                }
            }
            else
            {
                log.ErrorFormat("OnProcessMessage can't find: {0}", protoId);
                return false;
            }
            _manager.PushIoEvent(messageEvent);
            return true;
        }

        protected void PostSend(SocketAsyncEventArgs ioEventArgs)
        {
            var dataToken = (DataToken)ioEventArgs.UserToken;
            ioEventArgs.SetBuffer(dataToken.BufferOffset + dataToken.MessageDoneNum, dataToken.MessageLength - dataToken.MessageDoneNum);

            bool willRaiseEvent = ioEventArgs.AcceptSocket.SendAsync(ioEventArgs);
            if (!willRaiseEvent)
            {
                ProcessSend(ioEventArgs);
            }
        }

        protected void ProcessSend(SocketAsyncEventArgs ioEventArgs)
        {
            var dataToken = (DataToken)ioEventArgs.UserToken;
            if (ioEventArgs.SocketError == SocketError.Success)
            {
                dataToken.Reset(true);
                try
                {
                    TryDequeueAndPostSend(dataToken.Session, ioEventArgs);
                }
                catch (Exception ex)
                {
                    dataToken.Session.ResetSendFlag();
                    ReleaseIOEventArgs(ioEventArgs);
                    log.DebugFormat("ProcessSend error : {0}", ex.Message);
                    throw;
                }
            }
            else
            {
                log.WarnFormat("ProcessSend ioEventArgs.SocketError: {0} {1}", ioEventArgs.SocketError, dataToken.BufferOffset);
                dataToken.Session.ResetSendFlag();

                _manager.PushIoEvent(new ConnectionEventArgs()
                {
                    EventType = EnumConnectionEventType.Disconnect,
                    Session = dataToken.Session,
                    NodeId = _nodeId,
                });

                CloseClientSocket(ioEventArgs);
            }
        }

        protected void TryDequeueAndPostSend(NetSession session, SocketAsyncEventArgs ioEventArgs)
        {
            if (null == ioEventArgs)
            {
                log.Error("TryDequeueAndPostSend. ioEventArgs == null");
                return;
            }

            if (session.PendingSendDataLen() > 64 * 1024 * 100)
            {
                log.Warn($"thread:{Thread.CurrentThread.ManagedThreadId} threadpool:{Thread.CurrentThread.IsThreadPoolThread}. pendingSendDataLen:{session.PendingSendDataLen()}");
            }

            if (session.WorkSocket == null || !session.WorkSocket.Connected)
            {
                log.WarnFormat("TryDequeueAndPostSend socket is disconnetc: IP: {0} Hashcode:{1} ", session.RemoteEndPoint.ToString(), session.HashCode);
                ResetSAEAObject(ioEventArgs);
                ReleaseIOEventArgs(ioEventArgs);
                return;
            }

            var dataToken = (DataToken)ioEventArgs.UserToken;
            dataToken.Session = session;

            //将session中的待发送数据，读取到socket的缓存中，准备发送
            dataToken.MessageLength = session.ReadAndResetSendFlag(ioEventArgs.Buffer, dataToken.BufferOffset, _bufferManager.BufferSizeForEachBlock);

            if (dataToken.MessageLength > 0)
            {
                try
                {
                    PostSend(ioEventArgs);
                }
                catch (Exception ex)
                {
                    ReleaseIOEventArgs(ioEventArgs);
                    session.ResetSendFlag();
                    log.ErrorFormat("TryDequeueAndPostSend failed IP: {0} Hashcode:{1} Error:{2}", session.RemoteEndPoint.ToString(), session.HashCode, ex.Message.ToString());
                }
            }
            else
            {
                var isPendingData = false;
                lock (session.SocketSyncLock)
                {
                    if (session.PendingSendDataLen() > 0)
                    {
                        isPendingData = true;
                    }
                    else
                    {
                        session.ResetSendFlag();
                        ReleaseIOEventArgs(ioEventArgs);
                    }
                }

                if (isPendingData)
                {
                    TryDequeueAndPostSend(session, ioEventArgs);
                }
            }
        }

        public virtual void OnProtocolError()
        {
        }

        public EnumSendProtocolReturn SendProtocol(NetSession session, Pb.IProtocol proto)
        {
            var bytes = proto.Serialize();
            SendMessageAsync(session, bytes);
            return EnumSendProtocolReturn.OK;
        }

        public EnumSendProtocolReturn SendProtocol(long remotenodeId, Pb.IProtocol proto)
        {
            if(dictIdSession.ContainsKey(remotenodeId))
            {
                return SendProtocol(dictIdSession[remotenodeId], proto);
            }
            {
                return EnumSendProtocolReturn.Invalid_Session;
            }
        }

        public EnumSendProtocolReturn SendProtocol(Guid sessionId, Pb.IProtocol proto)
        {
            if(dictSession.ContainsKey(sessionId))
            {
                return SendProtocol(dictSession[sessionId], proto);
            }
            else
            {
                return EnumSendProtocolReturn.Invalid_Session;
            }
        }

    }
}
